package com.jiayuan.cn.energy.test;

//import ch.qos.logback.core.subst.Tokenizer;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.io.File;
import java.util.Arrays;


public class AppFlink {

    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
//            Arrays.stream(value.toLowerCase().split("\\W+")).filter(word -> !word.isEmpty()).forEach(word -> out.collect(new Tuple2<>(word, 1)));
            Arrays.stream(value.toLowerCase().split("\\s")).filter(word -> !word.isEmpty()).forEach(word -> out.collect(new Tuple2<>(word, 1)));
        }
    }

    public static void main(String[] args) throws Exception {
        String filePath = "count";
        File folder = new File(filePath);
        if (folder.exists()) {
            folder.delete();
//            .deleteOnExit();
        }
        // Define a configuration
        Configuration configuration = new Configuration();
//        configuration.setInteger(RestOptions.PORT, 8082);
        configuration.setString(RestOptions.BIND_PORT, "8088");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        DataStreamSource<String> text = env.fromData("Hello flink", "Hello world");
        DataStream<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).keyBy(0).sum(1);
        System.out.println("----------------------------------------------------");
        counts.print();
        counts.writeAsText(filePath);
//        String filePath = "see";
//        counts.writeAsText(filePath);
        env.execute("Wordcount");

//        String filePath = "D:\\Projects\\ProjectJiayuan\\CN\\energy-storage-helm-data\\count";
//        counts.writeAsText(filePath);
//        StreamingFileSink<MockService> sink2 = StreamingFileSink
//                .forRowFormat(
//                        new Path(filePath),
//                        new SimpleStringEncoder<MockService>("UTF-8"))
//                /**
//                 * 设置桶分配政策
//                 * DateTimeBucketAssigner --默认的桶分配政策，默认基于时间的分配器，每小时产生一个桶，格式如下yyyy-MM-dd--HH
//                 * BasePathBucketAssigner ：将所有部分文件（part file）存储在基本路径中的分配器（单个全局桶）
//                 */
//                .withBucketAssigner(new DateTimeBucketAssigner<>())
//                /**
//                 * 有三种滚动政策
//                 *  CheckpointRollingPolicy
//                 *  DefaultRollingPolicy
//                 *  OnCheckpointRollingPolicy
//                 */
//                .withRollingPolicy(
//                        /**
//                         * 滚动策略决定了写出文件的状态变化过程
//                         * 1. In-progress ：当前文件正在写入中
//                         * 2. Pending ：当处于 In-progress 状态的文件关闭（closed）了，就变为 Pending 状态
//                         * 3. Finished ：在成功的 Checkpoint 后，Pending 状态将变为 Finished 状态
//                         *
//                         * 观察到的现象
//                         * 1.会根据本地时间和时区，先创建桶目录
//                         * 2.文件名称规则：part-<subtaskIndex>-<partFileIndex>
//                         * 3.在macos中默认不显示隐藏文件，需要显示隐藏文件才能看到处于In-progress和Pending状态的文件，因为文件是按照.开头命名的
//                         *
//                         */
//                        DefaultRollingPolicy.builder()
//                                //每隔多久（指定）时间生成一个新文件
//                                .withRolloverInterval(TimeUnit.SECONDS.toMillis(2))
//                                //数据不活动时间 每隔多久（指定）未来活动数据，则将上一段时间（无数据时间段）也生成一个文件
//                                .withInactivityInterval(TimeUnit.SECONDS.toMillis(1))
//                                //每个文件大小
//                                .withMaxPartSize(1024 * 1024 * 1024)
//                                .build())
//                /**
//                 * 设置sink的前缀和后缀
//                 * 文件的头和文件扩展名
//                 */
//                .withOutputFileConfig(OutputFileConfig
//                        .builder()
//                        .withPartPrefix("lei")
//                        .withPartSuffix(".txt")
//                        .build())
//                .build();

    }


}